package com.atguigu.edu.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

public class DwdTrafficUserJumpDetail extends BaseApp {
    public static void main(String[] args) {
        new DwdTrafficUserJumpDetail().start(
                11102,
                4,
                "dwd_traffic_user_jump_detail",
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //kafkaStrDS.print("page");
        //将jsonStr转化成jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String value) throws Exception {
                        try {
                            return JSON.parseObject(value);
                        } catch (Exception e) {
                            throw new RuntimeException("不是一个完整的json");
                        }

                    }
                }
        );
        //指定事件时间字段和水位线
        SingleOutputStreamOperator<JSONObject> withTimestampsAndWatermarks = jsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                (element, recordTimestamp) -> element.getLong("ts")
                        )
        );
        //按照mid对流中的数据进行分组
        KeyedStream<JSONObject, String> keyByDS =
                withTimestampsAndWatermarks.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

        //定义cep匹配规则
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("first").where(
                        new IterativeCondition<JSONObject>() {
                            @Override
                            public boolean filter(JSONObject jsonObject, Context<JSONObject> context) throws Exception {
                                //会话的开头，last_page_id为空
                                String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                                return lastPageId == null;

                            }
                        }
                ).next("second")
                .where(new IterativeCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject, Context<JSONObject> context) throws Exception {
                        String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                        return lastPageId == null;
                    }
                }).within(Time.seconds(10L));
        PatternStream<JSONObject> patternStream = CEP.pattern(keyByDS, pattern);

        //提取匹配数据和超时数据
        OutputTag<String> timeoutTag = new OutputTag<String>("timeoutTag"){};
        SingleOutputStreamOperator<String> flatSelectStream = patternStream.flatSelect(timeoutTag, new PatternFlatTimeoutFunction<JSONObject, String>() {
            @Override
            public void timeout(Map<String, List<JSONObject>> pattern, long timeoutTimestamp, Collector<String> out) throws Exception {
                JSONObject first = pattern.get("first").get(0);
                out.collect(first.toJSONString());
            }
        }, new PatternFlatSelectFunction<JSONObject, String>() {
            @Override
            public void flatSelect(Map<String, List<JSONObject>> pattern, Collector<String> out) throws Exception {
                JSONObject first = pattern.get("first").get(0);
                out.collect(first.toJSONString());
            }
        });

        SideOutputDataStream<String> timeoutStream = flatSelectStream.getSideOutput(timeoutTag);

        //合并数据写出到kafka
        DataStream<String> unionStream = flatSelectStream.union(timeoutStream);
        unionStream.print("unionStream");
        unionStream.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_USER_JUMP_DETAIL));

        //执行任务
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
